package com.esreport.task;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.alibaba.fastjson.serializer.ValueFilter;
import com.amazonaws.services.kinesis.producer.*;
import com.esreport.Utils.StringUtil;
import com.esreport.dao.ReportRepository;
import com.esreport.entity.Report;
import com.esreport.logstash.tracker.FileTracker;
import com.google.common.collect.Iterables;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.UnsupportedEncodingException;
import java.nio.ByteBuffer;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author xuweiwei
 * @date 2018/12/29
 */
@Component
@Slf4j
public class LogStashTask implements InitializingBean, Runnable {
    @Autowired
    private ReportRepository reportRepository;


    @Autowired
    private FileTracker<Report> tracker;

    private KinesisProducer kinesisProducer;
    private ExecutorService executorService = Executors.newFixedThreadPool(3);

    @Override
    public void afterPropertiesSet() throws Exception {
        //初始化kinesis
        //=======================================================================
        String regionName = "cn-northwest-1";
        KinesisProducerConfiguration config = new KinesisProducerConfiguration()
                .setTempDirectory("/tmp/kinesis")
                .setRecordMaxBufferedTime(3000)
                .setMaxConnections(10)
                .setKinesisEndpoint("kinesis.cn-northwest-1.amazonaws.com.cn")
                .setRequestTimeout(60000)
                .setRegion(regionName);

        kinesisProducer = new KinesisProducer(config);
        //=======================================================================

        Thread thread = new Thread(this);
        thread.setName("LogStashTask");
        thread.start();
    }

    private static SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
    public String toJson(Object result) {
        return JSON.toJSONString(result,new ValueFilter(){
            @Override
            public Object process(Object o, String s, Object o1) {
                if(o1 instanceof Date) {
                    return sdf.format(o1);
                }
                return o1;
            }
        }, SerializerFeature.WriteNullNumberAsZero, SerializerFeature.WriteNullStringAsEmpty, SerializerFeature.WriteNullListAsEmpty, SerializerFeature.DisableCircularReferenceDetect);
    }

    private void addUserRecord(List<Report> reportList) throws UnsupportedEncodingException {
        for (Report report : reportList) {
            String json = toJson(report)+"\n";
            ByteBuffer data = ByteBuffer.wrap(json.getBytes("UTF-8"));
            ListenableFuture<UserRecordResult> future = kinesisProducer.addUserRecord("reportlog", "qyhl", data);

            Futures.addCallback(future, new FutureCallback<UserRecordResult>() {
                @Override
                public void onFailure(Throwable t) {
                    if (t instanceof UserRecordFailedException) {
                        Attempt last = Iterables.getLast(((UserRecordFailedException) t).getResult().getAttempts());
                        log.error(String.format("Record failed to put - %s : %s", last.getErrorCode(), last.getErrorMessage()));
                    }
                    log.error("Exception during put" + t);
                }

                @Override
                public void onSuccess(UserRecordResult result) {
                    log.info(" aws kinesis data streams  write success,sequenceNumber:{}", result.getSequenceNumber());
                }
            },executorService );
        }

    }

    @Override
    public void run() {
        log.info("索引更新模块启动");
        Queue<Report> queue = tracker.getQueue();
        List<Report> reportList = new ArrayList<Report>();
        while (true) {
            try {
                Report report = queue.poll();
                if (report == null) {
                    Thread.sleep(100);
                    continue;
                }
                if(StringUtil.isEmpty(report.getAction()) || StringUtil.isEmpty(report.getDeviceId())){
                    continue;
                }
                reportList.add(report);

                //记录超过100个,或者达到1分钟就刷新
                if (reportList.size() >= 10 ) {
                    log.info("更新索引");
                    try {
                        reportRepository.saveAll(reportList);
                    }catch (Exception e){
                        log.error("write es  error",e);
                    }
                    //写入到 aws kinesis  data streams
                    try {
                        addUserRecord(reportList);
                    }catch (Exception e){
                        log.error("write aws kinesis data stream error",e);
                    }
                    reportList.clear();
                    Thread.sleep(1*1000);
                }
            } catch (Exception e) {
                log.error("更新索引文件发生异常", e);
            }
        }
    }
}
